package org.jboss.netty.channel.socket.oio;

import static org.jboss.netty.channel.Channels.fireChannelBound;
import static org.jboss.netty.channel.Channels.fireChannelConnected;
import static org.jboss.netty.channel.Channels.fireExceptionCaught;
import gnu.io.CommPortIdentifier;
import gnu.io.SerialPort;

import java.io.PushbackInputStream;
import java.net.SocketAddress;
import java.util.concurrent.Executor;

import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelState;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.util.ThreadRenamingRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;

import uqbar.comm.netty.ProgramException;

/**
 * Todavia no me queda clara la division de responsabilidades entre Sink y
 * Channel.
 * 
 * @author leo
 * 
 */
public class SerialClientSocketPipelineSink extends OioClientSocketPipelineSink {

	private final Executor workerExecutor;

	public SerialClientSocketPipelineSink(final Executor workerExecutor) {
		super(workerExecutor);
		this.workerExecutor = workerExecutor;
	}

	/**
	 * Este metodo lo tengo que sobreescribir porque el switch de la superclase
	 * invoca un metodo privado que no puedo sobreescribirlo :(
	 */

	@Override
	public void eventSunk(final ChannelPipeline pipeline, final ChannelEvent e) throws Exception {
		OioClientSocketChannel channel = (OioClientSocketChannel) e.getChannel();
		ChannelFuture future = e.getFuture();

		if (e instanceof ChannelStateEvent) {
			ChannelStateEvent stateEvent = (ChannelStateEvent) e;
			ChannelState state = stateEvent.getState();
			Object value = stateEvent.getValue();
			switch (state) {
			case BOUND:
				if (value != null) {
					this.bind(channel, future, (SocketAddress) value);
				} else {
					super.eventSunk(pipeline, e);
				}
				break;
			case CONNECTED:
				if (value != null) {
					this.connect(channel, future, (SocketAddress) value);
				} else {
					super.eventSunk(pipeline, e);
				}
				break;
			}
		} else {
			super.eventSunk(pipeline, e);
		}

	}

	/**
	 * Este metodo es practicamente una copia del que esta en la superclase, lo
	 * que pasa que no estaba preparado para ser sobreescrito por otro tipo de
	 * socket address.
	 * 
	 * @param channel
	 * @param future
	 * @param remoteAddress
	 */
	protected void connect(final OioClientSocketChannel channel, final ChannelFuture future,
			final SocketAddress remoteAddress) {

		boolean bound = channel.isBound();
		boolean connected = false;
		boolean workerStarted = false;

		future.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

		try {

			// channel.socket.connect(remoteAddress,
			// channel.getConfig().getConnectTimeoutMillis());

			// Obtain I/O stream.
			SerialClientSocketChannel serialChannel = (SerialClientSocketChannel) channel;
			this.connectPort(serialChannel, remoteAddress);
			connected = true;
			channel.in = new PushbackInputStream(serialChannel.createInputStream(), 1);
			channel.out = serialChannel.createOutpurStream();

			// Fire events.
			future.setSuccess();
			if (!bound) {
				fireChannelBound(channel, channel.getLocalAddress());
			}
			fireChannelConnected(channel, channel.getRemoteAddress());

			// Start the business.
			DeadLockProofWorker.start(this.workerExecutor, new ThreadRenamingRunnable(new OioWorker(channel),
					"Old I/O client worker (" + channel + ')'));
			// HACK para esperar al arduino
			Thread.sleep(2000);
			workerStarted = true;
		} catch (Throwable t) {
			future.setFailure(t);
			fireExceptionCaught(channel, t);
		} finally {
			if (connected && !workerStarted) {
				OioWorker.close(channel, future);
			}
		}
	}

	// TODO no se si esto funciona. me parece que tengo que dejar de escuchar el
	// puerto y volver a conectar
	protected void bind(final OioClientSocketChannel channel, final ChannelFuture future,
			final SocketAddress localAddress) {
		try {
			SerialClientSocketChannel serialChannel = (SerialClientSocketChannel) channel;
			serialChannel.setAddress(localAddress);
			future.setSuccess();
			fireChannelBound(channel, channel.getLocalAddress());
		} catch (Throwable t) {
			future.setFailure(t);
			fireExceptionCaught(channel, t);
		}
	}

	public void connectPort(final SerialClientSocketChannel serialChannel, final SocketAddress remoteAddress) {
		try {
			SerialPortSocketAddress portAddress = (SerialPortSocketAddress) remoteAddress;
			serialChannel.setAddress(remoteAddress);
			serialChannel.setPortIdentifier(CommPortIdentifier.getPortIdentifier(portAddress.getPort()));
			serialChannel.setSerialPort((SerialPort) serialChannel.getPortIdentifier().open(
					serialChannel.getClass().getName(), serialChannel.getTimeOut()));

			// set port parameters
			serialChannel.getSerialPort().setSerialPortParams(serialChannel.getDataRate(),
					gnu.io.SerialPort.DATABITS_8, gnu.io.SerialPort.STOPBITS_1, gnu.io.SerialPort.PARITY_NONE);
			serialChannel.setConnected(true);
		} catch (Exception e) {
			throw ProgramException.wrap(e);
		}
	}

}
